package com.azzcs.myrpc.register;

import com.azzcs.myrpc.client.RpcClient;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.util.CollectionUtils;

import java.util.*;

/**
 * @Author: wzg
 * @Date: 2021/1/29 下午4:02
 */
public class ZookeeperServerRegister implements ServerRegister {

    private static final String SEPARATOR = "#";
    private static final long INTERVAL = 5L * 1000;

    private CuratorFramework client;

    public ZookeeperServerRegister(String registryAddress) {
        ExponentialBackoffRetry retry = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .retryPolicy(retry)
                .namespace("myrpc-server")
                .connectionTimeoutMs(30000)
                .sessionTimeoutMs(50000)
                .connectString(registryAddress)
                .build();
        client.start();
        this.client = client;
    }

    @Override
    public void register(String serverName, String address) {
        try {
            String path = getPath(serverName, address);
            Stat stat = client.checkExists().forPath(path);
            if (stat == null) {
                client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL)
                        .forPath(path, defaultData());
                System.out.println(serverName + "服务注册成功,地址为：" + address);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private byte[] defaultData() {
        return (System.currentTimeMillis() + SEPARATOR + 0).getBytes();
    }

    @Override
    public List<String> getServerAddress(String serverName) {
        List<String> addressList = new ArrayList<>();
        try {
            String path = getPath(serverName);
            Stat stat = client.checkExists().forPath(path);
            if (stat != null) {
                addressList = client.getChildren().forPath(path);
            } else {
                throw new RuntimeException("没有可用的服务端");
            }
        } catch (Exception e) {
            System.out.println(serverName + "服务不可用");
        }
        return addressList;
    }

    @Override
    public void reportTime(String serverName, String address, long useTime) {
        try {
            long currentTime = System.currentTimeMillis();
            client.setData().forPath(getPath(serverName, address), (currentTime + SEPARATOR + useTime).getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public String loadBalance(String serverName) {
        List<String> serverAddress = getServerAddress(serverName);
        if (CollectionUtils.isEmpty(serverAddress)) {
            return null;
        }
        String address = null;
        long minResponseTime = Long.MAX_VALUE;

        for (int i = 0; i < serverAddress.size(); i++) {
            ResponseTime time = getResponseTime(serverName, serverAddress.get(i));
            if (time == null) {
                continue;
            }
            if (time.latestResponseTime == 0) {
                address = serverAddress.get(i);
                minResponseTime = 0L;
                break;
            }
            if (time.latestResponseTime < minResponseTime) {
                minResponseTime = time.latestResponseTime;
                address = serverAddress.get(i);
            }
        }
        System.out.println("负载均衡服务器地址为：" + address + ",5秒内上一次响应时间为：" + minResponseTime);

        return address;
    }

    private ResponseTime getResponseTime(String serverName, String address) {
        try {
            String path = getPath(serverName, address);
            byte[] bytes = client.getData().forPath(path);
            String data = new String(bytes);
            String[] split = data.split(SEPARATOR);
            long latestExecuteTime = Long.parseLong(split[0]);
            long latestResponseTime = Long.parseLong(split[1]);
            return new ResponseTime(latestExecuteTime, latestResponseTime);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    private class ResponseTime {
        public long latestExecuteTime;
        public long latestResponseTime;

        public ResponseTime(long latestExecuteTime, long latestResponseTime) {
            this.latestExecuteTime = latestExecuteTime;
            this.latestResponseTime = latestResponseTime;
        }
    }

    private String getPath(String serverName, String address) {
        return "/" + serverName + "/" + address;
    }

    private String getPath(String serverName) {
        return "/" + serverName;
    }

    @Override
    public void registerListener(RpcClient rpcClient, String serverName) {
        try {
            String path = getPath(serverName);
            PathChildrenCache nodeCache = new PathChildrenCache(client, path, true);
            nodeCache.start();
            nodeCache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                    List<String> address = curatorFramework.getChildren().forPath(path);
                    rpcClient.updateServerAddress(serverName, address);
                }

            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void scheduleReport(String serverName) {
        try {
            String path = getPath(serverName);
            Stat stat = client.checkExists().forPath(path);
            if (stat == null) {
                return;
            }
            List<String> addressList = client.getChildren().forPath(path);
            for (String addr : addressList) {
                ZookeeperServerRegister.ResponseTime time = getResponseTime(serverName, addr);
                long latestExecuteTime = time.latestExecuteTime;
                if (System.currentTimeMillis() - latestExecuteTime - INTERVAL > 0) {
                    client.setData().forPath(getPath(serverName, addr), defaultData());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    @Override
    public void remove(String serviceName, String address) {
        try {
            client.delete().forPath(getPath(serviceName,address));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
